/**
 * Copyright (C) 2015 kmy-framework-base Project
 *               Author: Chris
 *               Date: 2015年11月7日
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.zmq.ha.client.core;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.zmq.ha.client.declaration.IWorkerhook;

/**
 * Name: ExecuteConsumer.java
 * ProjectName: [kmy-framework-base]
 * Package: [com.kmy.framework.rabbitmq.recover.core.ExecuteConsumer.java]
 * Description: TODO  
 * 
 * @since JDK1.7
 * @see
 *
 * @author: Chris
 * @date: 2015年11月7日 下午4:04:50
 *
 * Update-User: @author
 * Update-Time:
 * Update-Remark:
 * 
 * Check-User:
 * Check-Time:
 * Check-Remark:
 * 
 * Company: kmhc
 * Copyright: kmhc
 */
public class ExecuteConsumer extends DefaultConsumer {
	
	
	final Channel channel;
	
	final IWorkerhook hook;
	final String exchange;
	final String severity;
	final String queueName;
	final String consumerName;
	//区分是否由框架自动ACK，如果是，则不理会后续操作是否完成，只要调用了就认为是投递成功。
	final boolean isAutoAck;

	/** 
	 * @Title: ExecuteConsumer
	 * @Description: TODO
	 * @param @param channel         
	 * @throws 
	 */ 
	
	public ExecuteConsumer(Channel channel, IWorkerhook workerhook,String exchange,
			String severity, String queueName, String consumerName, boolean isAutoAck ) {
		super(channel);
		
		this.channel = channel;
		this.hook = workerhook;
		this.exchange =  exchange;
		this.severity = severity;
		this.queueName = queueName;
		this.consumerName = consumerName;
		this.isAutoAck = isAutoAck;
	}

	@Override
	public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
			throws IOException {
		
		try {
			long deliveryTag = envelope.getDeliveryTag();
			
			if( isAutoAck ){
				//钩子，用于调用外部的执行代码
				hook.receive(body);
				channel.basicAck(deliveryTag, false);
			} else{
				hook.receiveAndSelfAck(body, deliveryTag, channel);
			}
			
		} catch (Exception e) {
			e.printStackTrace();
			throw e;
		}
	}
	
	public void startReceiveMessage() throws IOException{
		try {
			channel.exchangeDeclare(this.exchange, "direct", true);
			channel.queueDeclare(this.queueName, true, false, false, null);
			channel.queueBind(this.queueName, this.exchange, this.severity);
			channel.basicQos(1);
			channel.basicConsume(this.queueName, false, consumerName, this);
		} catch (IOException e) {
			e.printStackTrace();
			throw e;
		}
	}
	
	public void stopConsumer(){
		try {
			this.channel.basicCancel(this.consumerName);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

}
